package faceoff.sample;

import static me.prettyprint.hector.api.factory.HFactory.*;

import java.util.ArrayList;
import java.util.List;

import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.beans.HColumn;
import me.prettyprint.hector.api.beans.HSuperColumn;
import me.prettyprint.hector.api.exceptions.HectorException;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.mutation.Mutator;
import me.prettyprint.hector.api.query.QueryResult;
import me.prettyprint.hector.api.query.SuperColumnQuery;

public class SimpleSuperCassandraDao {
	private String columnFamilyName;
	private Keyspace keyspace;
	private final StringSerializer serializer = StringSerializer.get();

	public HColumn<String, String> createColumn(String columnName, String value) {
		return HFactory.createColumn(columnName, value, serializer, serializer);
	}

	public void insertSuper(final String key, final String superColumnName, List<HColumn<String, String>> columns) {
		Mutator<String> m = createMutator(keyspace, serializer);
		HSuperColumn<String, String, String> hsc = HFactory.createSuperColumn(superColumnName, columns, serializer, serializer, serializer);
		m.insert(key, columnFamilyName, hsc);
	}

	public void insertSuper(final String key, final String superColumnName, final String columnName, final String value) {
		HColumn<String, String> newColumn = createColumn(columnName, value);

		SuperColumnQuery<String, String, String, String> q = HFactory.createSuperColumnQuery(keyspace, serializer, serializer, serializer, serializer);
		try {
			QueryResult<HSuperColumn<String, String, String>> r = q.setKey(key).setSuperName(superColumnName).setColumnFamily(columnFamilyName).execute();
			HSuperColumn<String, String, String> sc = r.get();
			sc.getColumns().add(newColumn);
		} catch (Exception e) {
			List<HColumn<String, String>> columns = new ArrayList<HColumn<String, String>>();
			columns.add(newColumn);
			insertSuper(key, superColumnName, columns);
		}
	}

	public String get(final String key, final String superColumnName, final String columnName) throws HectorException {
		SuperColumnQuery<String, String, String, String> q = HFactory.createSuperColumnQuery(keyspace, serializer, serializer, serializer, serializer);
		QueryResult<HSuperColumn<String, String, String>> r = q.setKey(key).setSuperName(superColumnName).setColumnFamily(columnFamilyName).execute();
		HSuperColumn<String, String, String> sc = r.get();

		if (sc.getColumns() == null) {
			return null;
		}

		for (HColumn<String, String> column : sc.getColumns()) {
			if (column.getName().equals(columnName)) {
				return column.getValue();
			}
		}
		return null;
	}

	public void setColumnFamilyName(String columnFamilyName) {
		this.columnFamilyName = columnFamilyName;
	}

	public void setKeyspace(Keyspace keyspace) {
		this.keyspace = keyspace;
	}
}
